package dbtask

import (
	dgcoll "github.com/darwinOrg/go-common/collection"
	dgctx "github.com/darwinOrg/go-common/context"
	dgerr "github.com/darwinOrg/go-common/enums/error"
	dglogger "github.com/darwinOrg/go-logger"
	"github.com/rolandhe/daog"
	"github.com/rolandhe/daog/ttypes"
	"time"
)

var CommonTaskExtDao = &commonTaskExtDao{}

type commonTaskExtDao struct {
}

func (d *commonTaskExtDao) InsertInitTask(tc *daog.TransContext, taskType string, channel string, content string) (int64, error) {
	now := time.Now()
	task := &CommonTask{
		Type:       taskType,
		Channel:    channel,
		Content:    *ttypes.FromString(content),
		Status:     TaskStatus.INIT,
		CreatedAt:  ttypes.NormalDatetime(now),
		ModifiedAt: ttypes.NormalDatetime(now),
	}

	_, err := CommonTaskDao.Insert(tc, task)
	if err != nil {
		return 0, err
	}
	if task.Id == 0 {
		return 0, dgerr.SYSTEM_ERROR
	}

	return task.Id, nil
}

func (d *commonTaskExtDao) InsertInitTaskWithBizId(tc *daog.TransContext, taskType, channel, content string, bizId int64) (int64, error) {
	now := time.Now()
	task := &CommonTask{
		Type:       taskType,
		Channel:    channel,
		Content:    *ttypes.FromString(content),
		Status:     TaskStatus.INIT,
		BizId:      bizId,
		CreatedAt:  ttypes.NormalDatetime(now),
		ModifiedAt: ttypes.NormalDatetime(now),
	}

	_, err := CommonTaskDao.Insert(tc, task)
	if err != nil {
		return 0, err
	}
	if task.Id == 0 {
		return 0, dgerr.SYSTEM_ERROR
	}

	return task.Id, nil
}

func (d *commonTaskExtDao) InsertInitDelayedTaskWithBizId(tc *daog.TransContext, taskType string, content string, bizId int64, delaySeconds int) (int64, error) {
	now := time.Now()
	task := &CommonTask{
		Type:             taskType,
		Content:          *ttypes.FromString(content),
		Status:           TaskStatus.INIT,
		BizId:            bizId,
		ScheduledStartAt: *ttypes.FromDatetime(now.Add(time.Duration(delaySeconds) * time.Second)),
		CreatedAt:        ttypes.NormalDatetime(now),
		ModifiedAt:       ttypes.NormalDatetime(now),
	}
	_, err := CommonTaskDao.Insert(tc, task)
	if err != nil {
		return 0, err
	}
	if task.Id == 0 {
		return 0, dgerr.SYSTEM_ERROR
	}
	return task.Id, nil
}

func (d *commonTaskExtDao) LockTasksToProcessing(tc *daog.TransContext, req *FetchTaskRequest) ([]*CommonTask, error) {
	matcher := daog.NewMatcher().
		Eq(CommonTaskFields.Type, req.TaskType).
		In(CommonTaskFields.Status, daog.ConvertToAnySlice(ToHandleStatuses))
	if req.Channel != "" {
		matcher.Eq(CommonTaskFields.Channel, req.Channel)
	}
	if req.FixedLockedBy {
		matcher.Add(daog.NewOrMatcher().Null(CommonTaskFields.LockedBy, false).Eq(CommonTaskFields.LockedBy, req.LockedBy))
	}
	if req.IsDelayedTask {
		matcher.Null(CommonTaskFields.ScheduledStartAt, true).
			Lte(CommonTaskFields.ScheduledStartAt, time.Now())
	}
	pager := daog.NewPager(req.PageSize, 1)
	order := daog.NewOrder(CommonTaskFields.CreatedAt)
	tasks, err := CommonTaskDao.QueryPageListMatcherForUpdate(tc, matcher, pager, true, order)
	if err != nil {
		return nil, err
	}
	if len(tasks) == 0 {
		return []*CommonTask{}, nil
	}

	now := time.Now()
	lockUntil := now.Add(time.Millisecond * time.Duration(req.LockMilli))

	modifier := daog.NewModifier().
		Add(CommonTaskFields.Status, TaskStatus.PROCESSING).
		Add(CommonTaskFields.StartAt, ttypes.NormalDatetime(now)).
		Add(CommonTaskFields.LockedAt, ttypes.NormalDatetime(now)).
		Add(CommonTaskFields.LockUntil, lockUntil).
		Add(CommonTaskFields.LockedBy, req.LockedBy).
		Add(CommonTaskFields.ModifiedAt, ttypes.NormalDatetime(now))
	_, err = CommonTaskDao.UpdateByIds(tc, modifier, dgcoll.MapToList(tasks, func(task *CommonTask) int64 { return task.Id }))
	if err != nil {
		return nil, err
	}

	return tasks, nil
}

func (d *commonTaskExtDao) EndAsSuccess(tc *daog.TransContext, taskId int64) error {
	now := time.Now()
	modifier := daog.NewModifier().
		Add(CommonTaskFields.Status, TaskStatus.SUCCESS).
		Add(CommonTaskFields.EndAt, ttypes.NormalDatetime(now)).
		SelfAdd(CommonTaskFields.ProcessedCount, 1).
		Add(CommonTaskFields.ModifiedAt, ttypes.NormalDatetime(now))
	_, err := CommonTaskDao.UpdateById(tc, modifier, taskId)
	return err
}

func (d *commonTaskExtDao) ResetProcessedCount(tc *daog.TransContext, taskId int64) error {
	now := time.Now()
	modifier := daog.NewModifier().
		Add(CommonTaskFields.ProcessedCount, 0).
		Add(CommonTaskFields.ModifiedAt, ttypes.NormalDatetime(now))
	_, err := CommonTaskDao.UpdateById(tc, modifier, taskId)
	return err
}

func (d *commonTaskExtDao) EndAsFail(tc *daog.TransContext, taskId int64, reason string) error {
	now := time.Now()
	modifier := daog.NewModifier().
		Add(CommonTaskFields.Status, TaskStatus.FAIL).
		Add(CommonTaskFields.EndAt, ttypes.NormalDatetime(now)).
		Add(CommonTaskFields.Reason, reason).
		SelfAdd(CommonTaskFields.ProcessedCount, 1).
		Add(CommonTaskFields.ModifiedAt, ttypes.NormalDatetime(now))
	_, err := CommonTaskDao.UpdateById(tc, modifier, taskId)
	return err
}

func (d *commonTaskExtDao) EndAsCanceled(tc *daog.TransContext, taskId int64, reason string) error {
	now := time.Now()
	modifier := daog.NewModifier().
		Add(CommonTaskFields.Status, TaskStatus.CANCELED).
		Add(CommonTaskFields.EndAt, ttypes.NormalDatetime(now)).
		Add(CommonTaskFields.Reason, reason).
		SelfAdd(CommonTaskFields.ProcessedCount, 1).
		Add(CommonTaskFields.ModifiedAt, ttypes.NormalDatetime(now))
	_, err := CommonTaskDao.UpdateById(tc, modifier, taskId)
	return err
}

func (d *commonTaskExtDao) ReInit(tc *daog.TransContext, taskId int64) error {
	now := time.Now()
	modifier := daog.NewModifier().
		Add(CommonTaskFields.Status, TaskStatus.INIT).
		SelfAdd(CommonTaskFields.ProcessedCount, 1).
		Add(CommonTaskFields.ModifiedAt, ttypes.NormalDatetime(now))
	_, err := CommonTaskDao.UpdateById(tc, modifier, taskId)
	return err
}

func (d *commonTaskExtDao) GetByIdsAndLockedBy(tc *daog.TransContext, taskIds []int64, lockedBy string) ([]*CommonTask, error) {
	return CommonTaskDao.QueryListMatcher(tc, daog.NewMatcher().
		In(CommonTaskFields.Id, daog.ConvertToAnySlice(taskIds)).
		Eq(CommonTaskFields.LockedBy, lockedBy))
}

func (d *commonTaskExtDao) ReInitTimeoutProcessingTasks(tc *daog.TransContext, taskType string, timeoutMinutes int) (int64, error) {
	now := time.Now()
	modifier := daog.NewModifier().
		Add(CommonTaskFields.Status, TaskStatus.INIT).
		Add(CommonTaskFields.Reason, "").
		Add(CommonTaskFields.StartAt, nil).
		Add(CommonTaskFields.EndAt, nil).
		Add(CommonTaskFields.ModifiedAt, ttypes.NormalDatetime(now))
	matcher := daog.NewMatcher().
		Eq(CommonTaskFields.Type, taskType).
		Eq(CommonTaskFields.Status, TaskStatus.PROCESSING).
		Lt(CommonTaskFields.StartAt, ttypes.NormalDatetime(now.Add(time.Minute*time.Duration(-timeoutMinutes))))
	return CommonTaskDao.UpdateByModifier(tc, modifier, matcher)
}

func (d *commonTaskExtDao) CancelByIds(tc *daog.TransContext, taskIds []int64, reason string) error {
	now := time.Now()
	modifier := daog.NewModifier().
		Add(CommonTaskFields.Status, TaskStatus.CANCELED).
		Add(CommonTaskFields.Reason, reason).
		Add(CommonTaskFields.ModifiedAt, ttypes.NormalDatetime(now))
	matcher := daog.NewMatcher().
		In(CommonTaskFields.Id, daog.ConvertToAnySlice(taskIds)).
		In(CommonTaskFields.Status, daog.ConvertToAnySlice(ToHandleStatuses))
	_, err := CommonTaskDao.UpdateByModifier(tc, modifier, matcher)
	return err
}

func (d *commonTaskExtDao) FilterUnFinishedTaskIds(tc *daog.TransContext, taskIds []int64) ([]int64, error) {
	matcher := daog.NewMatcher().
		In(CommonTaskFields.Id, daog.ConvertToAnySlice(taskIds)).
		In(CommonTaskFields.Status, daog.ConvertToAnySlice(UnFinishedStatuses))
	tasks, err := CommonTaskDao.QueryListMatcherWithViewColumns(tc, matcher, []string{CommonTaskFields.Id})
	if err != nil {
		return nil, err
	}
	return dgcoll.MapToList(tasks, func(task *CommonTask) int64 { return task.Id }), nil
}

func (d *commonTaskExtDao) DeleteOverdueTasks(ctx *dgctx.DgContext, tc *daog.TransContext, overdueDays int) error {
	matcher := daog.NewMatcher().Lt(CommonTaskFields.CreatedAt, time.Now().Add(time.Hour*24*time.Duration(-overdueDays)))
	count, err := CommonTaskDao.DeleteByMatcher(tc, matcher)
	if err != nil {
		dglogger.Errorf(ctx, "delete overdue tasks error: %v", err)
		return err
	}
	dglogger.Infof(ctx, "delete overdue tasks count: %d", count)
	return nil
}
